package cn.tecnova.test

import cn.tecnova.utils.ConfigHandler
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext

/**
  * description:
  * Rabcheng
  * Date:2019/5/30 11:03
  **/
object Financial_insurance_claims2es {

  Logger.getLogger("org").setLevel(Level.ERROR)
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
            .setMaster("local[*]")
      .set("es.index.auto.create", "true")
      .set("es.nodes", "192.168.100.3,192.168.100.6")
      .set("es.port", "9200")
      .set("es.nodes.wan.only", "true")

    val sc = new SparkContext(conf)
    val sqLContext = new SQLContext(sc)

    import sqLContext.implicits._
    val userSubjectDF = sqLContext.read.jdbc(ConfigHandler.url,"financial_insurance_claims",ConfigHandler.props)


    //导入写es相关的包
    import org.elasticsearch.spark.sql._

    val res2 = userSubjectDF.map(row=>{
      Financial_insurance_claims(
        row.getAs[String]("id"),
        row.getAs[Int]("incId"),
        row.getAs[String]("reportNumber"),
        row.getAs[String]("claimNumber"),
        row.getAs[String]("policyNo"),
        row.getAs[Int]("claimType"),
        row.getAs[String]("insCompanyId"),
        row.getAs[String]("policyHolder"),
        row.getAs[String]("policyHolderIdCard"),
        row.getAs[String]("policyHolderTel"),
        row.getAs[String]("insuredName"),
        row.getAs[String]("insuredTel"),
        row.getAs[String]("insuredIdCard"),
        row.getAs[String]("plateNumber"),
        row.getAs[String]("vinCode"),
        row.getAs[String]("ownerName"),
        row.getAs[String]("vehicleType"),
        row.getAs[Int]("vehicleFeatures"),
        row.getAs[Int]("vehicleNature"),
        row.getAs[Int]("isSecondCar"),
        row.getAs[Double]("newCarPrice"),
        row.getAs[Int]("isNotNew"),
        row.getAs[String]("alarmPerson"),
        row.getAs[String]("alarmTel"),
        row.getAs[String]("accidentAddr"),
        row.getAs[Int]("accidentType"),
        row.getAs[Int]("respDivision"),
        row.getAs[String]("surveyPerson"),
        row.getAs[String]("surveyPersonId"),
        row.getAs[String]("driverName"),
        row.getAs[String]("driverTel"),
        row.getAs[String]("driverNumber"),
        row.getAs[String]("threeNumber"),
        row.getAs[String]("injuryNumber"),
        row.getAs[String]("claimAgent"),
        row.getAs[String]("claimAgentIdCard"),
        row.getAs[String]("claimAgentTel"),
        row.getAs[String]("claimAgentBankcard"),
        row.getAs[Double]("carLoss"),
        row.getAs[String]("maintenanceUnit"),
        row.getAs[Int]("lossMethod"),
        row.getAs[Double]("reimburseAmount"),
        row.getAs[String]("contacts"),
        row.getAs[String]("contactsTel"),
        row.getAs[String]("briefCase"),
        row.getAs[String]("creator"),
        row.getAs[String]("modifier"),
        row.getAs[String]("carLevel"),
        row.getAs[Int]("finClaimsType"),
        row.getAs[String]("deptOrComName"),
        row.getAs[Int]("state"),
        row.getAs[String]("insuranceType")
      )
    })

    res2.saveToEs("financial_insurance_claims" + "/financial_insurance_claims")

    sc.stop()

  }

}
